package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/22
 * @Desc: 学大数据，上多易教育
 *    规则模型A的通用 运算机类
 *    它能支持类似如下示例的规则：
 *
 * 动态画像条件：  X行为 > 3次；  所以 W行为中的 p1属性的最大值 > p1属性的最小值 * 5
 * 触发事件：     Q行为且属性 p2 > 10

      规则模型A，应该遵循如下的参数模板：
               {
                   "rule_id":"rule_A_001",
                   "dynamic_profile_condition1": {
                       "event_id": "R",
                       "operator": ">",
                       "count": 6
                    },
                   "dynamic_profile_condition2": {
                       "event_id": "K",
                       "property_name": "p2",
                       "operator": ">",
                       "max_min_bs": 8
                    },
                   "trigger_condition": {
                       "event_id": "M",
                       "property_name": "p1",
                       "operator": ">",
                       "compare_value": 2

                    }
               }
 **/
public class RuleModel_A_Calculator implements RuleCalculator {
    ValueState<Integer> cntState;
    ValueState<Double> minState;
    ValueState<Double> maxState;

    JSONObject paramJsonObject;

    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext,String ruleParamJson) throws Exception {

        // 解析规则参数json
        paramJsonObject = JSON.parseObject(ruleParamJson);
        String ruleId = paramJsonObject.getString("rule_id");


        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cntState", Integer.class));
        minState = runtimeContext.getState(new ValueStateDescriptor<Double>("minState", Double.class));
        maxState = runtimeContext.getState(new ValueStateDescriptor<Double>("maxState", Double.class));

        message = new JSONObject();
        message.put("rule_id",ruleId);


    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {

        String eventId = userEvent.getEvent_id();
        long userId = userEvent.getUser_id();


        /**
         * 动态画像条件1的处理
         */
        // 取出动态画像条件1要求的eventId
        JSONObject dynamicProfileCondition1 = paramJsonObject.getJSONObject("dynamic_profile_condition1");
        String dynamicEventId_1 = dynamicProfileCondition1.getString("event_id");
        int dynamicCount_1 = dynamicProfileCondition1.getIntValue("count");

        if(dynamicEventId_1.equals(eventId)){
            cntState.update(cntState.value() == null? 1 : cntState.value()+1);
        }


        /**
         * 动态画像条件2的处理
         */
        JSONObject dynamicProfileCondition2 = paramJsonObject.getJSONObject("dynamic_profile_condition2");
        String dynamicEventId_2 = dynamicProfileCondition2.getString("event_id");
        String propertyName_2 = dynamicProfileCondition2.getString("property_name");
        String dynamicOperator_2 = dynamicProfileCondition2.getString("operator");
        int maxMinBs_2 = dynamicProfileCondition2.getInteger("max_min_bs");


        if(dynamicEventId_2.equals(eventId)){
            String p1ValueStr = userEvent.getProperties().get(propertyName_2);
            double p1Value = Double.parseDouble(p1ValueStr);

            if(minState.value() == null || minState.value()> p1Value){
                minState.update(p1Value);
            }

            if(maxState.value() == null || maxState.value() < p1Value){
                maxState.update(p1Value);
            }
        }


        /**
         * 触发事件的处理
         */
        JSONObject triggerCondition = paramJsonObject.getJSONObject("trigger_condition");
        String triggerEventId = triggerCondition.getString("event_id");
        String triggerPropertyName = triggerCondition.getString("property_name");
        String triggerOperator = triggerCondition.getString("operator");
        int triggerCompareValue = triggerCondition.getIntValue("compare_value");


        if(triggerEventId.equals(eventId)){

            // 判断触发条件中的属性要求是否满足
            boolean tmp = false;
            if(triggerOperator.equals(">")) {
                tmp = Double.parseDouble(userEvent.getProperties().get(triggerPropertyName)) > triggerCompareValue;
            }else if(triggerOperator.equals("=")){
                tmp = Double.parseDouble(userEvent.getProperties().get(triggerPropertyName)) == triggerCompareValue;
            }else if(triggerOperator.equals(">=")){
                tmp = Double.parseDouble(userEvent.getProperties().get(triggerPropertyName)) >= triggerCompareValue;
            }else if(triggerOperator.equals("<=")){
                tmp = Double.parseDouble(userEvent.getProperties().get(triggerPropertyName)) <= triggerCompareValue;
            }

            if(tmp) {
                if (cntState.value() != null && cntState.value() > dynamicCount_1
                        && minState != null & maxState != null
                        && maxState.value() > minState.value() *  maxMinBs_2  ) {

                    message.put("user_id", userId);
                    message.put("hit_time", userEvent.getAction_time());

                    out.collect(message.toJSONString());
                }
            }
        }

    }
}
